package com.hzh.SparkSQL

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.{DataFrame, SparkSession}

object Demo3DSLAPI {

  def main(args: Array[String]): Unit = {

    /**
     * 1、创建spark连接
     */
    val session: SparkSession = SparkSession
      .builder()
      .master("local")
      .appName("Demo3DslAPI")
      .config("spark.sql.shuffle.partition",1) //指定在shuffle之后的分区数，默认是200，类似hive中设置reduce的数量
      .getOrCreate()

    /**
     * DSL API
     */
    //读取一个json格式文件,spark会自动识别json中的列名
    val dataFrame: DataFrame = session
      .read
      .format("json") //指定读取数据的格式为json
      .load("data/students.json")

    dataFrame.printSchema()

    /**
     * show：查看df中数据，相当于rdd的action算子，会触发任务的执行
     *
     */

    dataFrame.show()
    //指定打印多少行
    dataFrame.show(20)
    //完整打印每一列的数据
    dataFrame.show(false)

    /**
     * select:选择数据，和sql中的select用法基本一致
     * dsl中select 不能使用聚合函数需要在agg中使用聚合函数
     * select相当于rdd中的转换算子
     * selectExpr可以传一个sql的表达式
     */
    dataFrame.select("id", "age")
    dataFrame.selectExpr("id", "age+1 as age").show()

    //需要导入spark sql的函数才能在DSl中使用函数
    import org.apache.spark.sql.functions._
    import session.implicits._
    //使用列对象的方式
    dataFrame.select($"name", $"age" + 1 as "age").show()
    //在sql中使用spark函数
    dataFrame.select($"id", substring($"clazz", 0, 2) as "zz").show()

    /**
     * where：过滤数据 也相当于一个转换算子
     */
    //字符串的sql表达式
    dataFrame.where("gender = '女' and age = 23").show()

    //使用列对象
    dataFrame.where($"gender" =!= "男" and $"age" === 22).show()
    //这个里面只能用函数不能写scala代码
    dataFrame.where(substring($"clazz", 0, 2) === "文科").show()

    /**
     * groupBy agg ：分组聚合要一起使用不能单独使用
     * 分组聚合之后返回的DF只包含分组字段和聚合字段
     * 分组不能在select中聚合
     */
    dataFrame
      .groupBy($"clazz")
      .agg(count($"clazz") as "num", floor(avg($"age")) as "avg_age1", ceil(avg($"age")) as "avg_age2")
      .show()

    /**
     * orderBy : 排序
     *
     */
    dataFrame
      .groupBy($"clazz")
      .agg(count($"clazz") as "num")
      .orderBy($"num".desc)
      .show()

    /**
     * join:表关联
     */
    //读取分数表
    val scoreDF: DataFrame = session
      .read
      .format("csv")
      .option("sep", ",")
      .schema("id STRING,cid STRING,Sco DOUBLE") //指定列名
      .load("data/score.txt")

    //关联字段名不一样时
    //    scoreDF.join(dataFrame,$"id"===$"sid","inner").show()
    //关联字段名一样时,直接写"id"
    val joinDF: DataFrame = scoreDF.join(dataFrame, "id")


    /**
     * 开窗函数
     * 统计每个班级总分前十的学生
     * withColumn():在DF的基础上增加新的列
     *
     */
    joinDF
      .groupBy($"id",$"clazz")//按照学号和班级分组
      .agg(sum($"sco") as "sumSco")//计算总分
      //简写，在前面的基础上增加列
      .withColumn("r",row_number() over Window.partitionBy($"clazz").orderBy($"sumSco".desc))
      .where($"r"<=10)
      .show()

    /**
     * sql
     *
     */
    joinDF.createOrReplaceTempView("student_score")

    session.sql(
      """
        |select * from (
        |select id as sid,clazz,sumSco,row_number() over(partition by clazz order by sumSco desc) as r from (
        |select id,clazz,sum(sco) as  sumSco
        |from student_score
        |group by id,clazz
        |) as a
        |) as b
        |where r<=10
        |""".stripMargin)

  }
}
